package framework.registry;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;

@Slf4j
public class ZookeeperRegistry implements Registry{

    private static final String framework_prefix = "/x-rpc";

    private final CuratorFramework client;

    public ZookeeperRegistry(String zkServerAddress) {
        this.client = CuratorFrameworkFactory.newClient(zkServerAddress, 1000 * 30, 1000 * 5, new ExponentialBackoffRetry(1000, 3, 1000));
        this.client.start();
        log.debug("zookeeper client start successful {}", zkServerAddress);
    }

    private String getServiceProviderPath(String interfaceName) {
        return framework_prefix + "/" + interfaceName + "/provider";
    }

    private String getInterfaceNodePath(String servicePath, ServiceAddress serviceAddress) {
        return servicePath + "/" + serviceAddress.getHost() + ":" + serviceAddress.getPort();
    }

    @Override
    public void serviceRegister(String interfaceName, ServiceAddress serviceAddress) {

        String serviceProviderPath = getServiceProviderPath(interfaceName);
        try {
            if (client.checkExists().forPath(serviceProviderPath) == null) {
                this.client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(serviceProviderPath);
            }
            String interfaceNodePath = getInterfaceNodePath(serviceProviderPath, serviceAddress);
            if (this.client.checkExists().forPath(interfaceNodePath) == null) {
                this.client.create().withMode(CreateMode.EPHEMERAL).forPath(interfaceNodePath);
            } else {
                this.client.setData().forPath(interfaceNodePath);
            }
            log.debug("registerService service [{}]", interfaceNodePath);
        } catch (Exception e) {
            log.error("registerService error [{}] [{}]", serviceProviderPath, serviceAddress.toJson());
            throw new RuntimeException(e);
        }
    }


    @Override
    public List<ServiceAddress> serviceDiscovery(String interfaceName) {

        String serviceProviderPath = getServiceProviderPath(interfaceName);
        try {
            if (this.client.checkExists().forPath(serviceProviderPath) != null) {
                List<String> services = this.client.getChildren().forPath(serviceProviderPath);
                List<ServiceAddress> addressList = services.stream()
                        .map(s -> s.split(":"))
                        .map(s2 -> new ServiceAddress(s2[0], Integer.parseInt(s2[1])))
                        .collect(Collectors.toList());
                log.debug("getService [{}]", interfaceName);
                for (ServiceAddress serviceAddress : addressList) {
                    log.debug(serviceAddress.toJson());
                }
                return addressList;
            }
        } catch (Exception e) {
            log.error("getService error [{}]", interfaceName);
            throw new RuntimeException(e);
        }

        return Collections.emptyList();
    }

    @Override
    public void subscribeService(String interfaceName, List<ServiceAddress> serviceAddresses) {

    }
}
